package com.bigdata.spark.mallapp_realtime.test

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object mall_log_test_app1 {

  def main(args: Array[String]): Unit = {
    val logtest: SparkConf = new SparkConf().setMaster("local[*]").setAppName("logtest")
    val ssc = new StreamingContext(logtest, Seconds(3))




    val KafkaPara = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "atguigu",
      "key.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer"

    )

    val KafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Set("gmall_event_0523"), KafkaPara))
    //不可以直接对kafkaDS进行print，因为kafkaDS

    KafkaDS.foreachRDD(

      iter=>{

        iter

      }




    )




    val res: DStream[String] = KafkaDS.map(
      kafkadata => {

        val str: String = kafkadata.value()
        print(str)
        str


      }
    )

    res.print()

    ssc.start()
    ssc.awaitTermination()
  }

}
